package com.boilermaker.mq.mqserver.manager;

import com.boilermaker.mq.common.MqException;
import com.boilermaker.mq.mqserver.core.Binding;
import com.boilermaker.mq.mqserver.core.Exchange;
import com.boilermaker.mq.mqserver.core.MSGQueue;
import com.boilermaker.mq.mqserver.core.Message;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/*
 * 统一管理内存中的全部数据
 * 需要注意该类的线程安全问题
 */
@Slf4j
public class MemoryDataCenter {
    // 管理交换机。
    // 结构：key-exchangeName val-Exchange
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();

    // 管理队列。
    // 结构：key-queueName val-MSGQueue
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();

    // 管理绑定关系（通过 exchangeName 快速查找所有与之相绑定的队列）
    // 结构：key-exchangeName val-<key-queueName val-Binding>
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> exchangeBindingMap = new ConcurrentHashMap<>();

    // 管理绑定关系（通过 queueName 快速查找所有与之相绑定的交换机）
    // 结构：key-queueName val-<key-exchangeName val-Binding>
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> queueBindingMap = new ConcurrentHashMap<>();

    // 消息中心（全部消息记录于此）
    // 结构：key-messageId val-Message
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();

    // 未取走的消息（记录消息从属于哪个队列，直到消息被消费者取走）
    // 结构：key-queueName val-messageList
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();

    // [用于 Ack 模式] 未被确认的消息（该模式下，消费者取走消息后需要显式通知服务器，表示已处理）
    // 结构：key-queueName val-<key-messageId val-Message>
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> waitAckMap = new ConcurrentHashMap<>();

    /*
     * 封装交换机操作
     */

    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        log.info("新交换机添加成功, exchangeName=" + exchange.getName());
    }

    public Exchange getExchange(String name) {
        return exchangeMap.get(name);
    }

    public void deleteExchange(String name) {
        exchangeMap.remove(name);
        log.info("已删除指定交换机, exchangeName=" + name);
    }

    /*
     * 封装队列操作
     */

    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
        log.info("新队列添加成功, queueName=" + queue.getName());
    }

    public MSGQueue getQueue(String name) {
        return queueMap.get(name);
    }

    public void deleteQueue(String name) {
        queueMap.remove(name);
        log.info("已删除指定队列, queueName=" + name);
    }

    /*
     * 封装绑定操作
     */

    public void insertBinding(Binding binding) {
        // 若此时内层的哈希表尚为空，computeIfAbsent 会自动创建，并放入外层哈希表的相应位置
        // computeIfAbsent 是 ConcurrentHashMap 提供的，因此它也是线程安全的
        // 分别维护两个 map
        ConcurrentHashMap<String, Binding> queueMap = exchangeBindingMap.computeIfAbsent(binding.getExchangeName(),
                k -> new ConcurrentHashMap<>());
        ConcurrentHashMap<String, Binding> exchangeMap = queueBindingMap.computeIfAbsent(binding.getQueueName(),
                k -> new ConcurrentHashMap<>());

        queueMap.put(binding.getQueueName(), binding);
        exchangeMap.put(binding.getExchangeName(), binding);
    }

    // 通过 exchangeName 和 queueName 确定唯一的绑定关系
    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> tempMap = exchangeBindingMap.get(exchangeName);
        if (tempMap == null) {
            return null;
        }
        return tempMap.get(queueName);
    }

    // 通过 exchangeName 找到该交换机的全部绑定关系
    public ConcurrentHashMap<String, Binding> getExchangeBindings(String exchangeName) {
        return exchangeBindingMap.get(exchangeName);
    }

    // 通过 queueName 找到该队列的全部绑定关系
    public ConcurrentHashMap<String, Binding> getQueueBindings(String queueName) {
        return queueBindingMap.get(queueName);
    }

    public void deleteBinding(Binding binding) throws MqException {
        if (getBinding(binding.getExchangeName(), binding.getQueueName()) == null) {
            throw new MqException("该绑定不存在，exchangeName=" + binding.getExchangeName()
                    + "，queueName=" + binding.getQueueName());
        }

        exchangeBindingMap.get(binding.getExchangeName()).remove(binding.getQueueName());
        queueBindingMap.get(binding.getQueueName()).remove(binding.getExchangeName());
    }

    /*
     * 封装消息操作
     */

    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        log.info("新消息添加成功, messageId=" + message.getMessageId());
    }

    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    public void deleteMessage(String messageId) {
        messageMap.remove(messageId);
        log.info("已删除指定消息, messageId=" + messageId);
    }

    // 发送消息到指定队列
    public void sendMessage(MSGQueue queue, Message message) {
        LinkedList<Message> messageList = queueMessageMap.computeIfAbsent(queue.getName(),
                k -> new LinkedList<>());
        synchronized (messageList) {
            messageList.add(message);
        }
        log.info("消息已被投递至指定队列中, queueName=" + queue.getName() + ", messageId=" + message.getMessageId());
        // 此时不清楚该 message 是否已经存在于消息中心，为保险起见再次调用 addMessage
        messageMap.put(message.getMessageId(), message);
    }

    // 从指定队列中取出顶部的消息
    public Message pollMessage(String queueName) {
        LinkedList<Message> messageList = queueMessageMap.get(queueName);
        if (messageList == null || messageList.isEmpty()) return null;
        synchronized (messageList) {
            Message message = messageList.remove(0);
            log.info("已取出消息, queueName=" + queueName + ", messageId=" + message.getMessageId());
            return message;
        }
    }

    // 获取指定队列中消息的个数
    public int getMessageCount(String queueName) {
        LinkedList<Message> messageList = queueMessageMap.get(queueName);
        if (messageList == null) return 0;
        return messageList.size();
    }

    // 添加未被确认的消息
    public void addMessageWaitAck(String queueName, Message message) {
        ConcurrentHashMap<String, Message> tempMap = waitAckMap.computeIfAbsent(queueName,
                k -> new ConcurrentHashMap<>());
        tempMap.put(message.getMessageId(), message);
        log.info("消息进入待确认队列, messageId=" + message.getMessageId());
    }

    // 获取指定的未确认消息
    public Message getMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> tempMap = waitAckMap.get(queueName);
        if (tempMap == null) return null;
        return tempMap.get(messageId);
    }

    // 将消息从待确认队列中移除（消息已被确认）
    public void removeMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> tempMap = waitAckMap.get(queueName);
        if (tempMap == null) return;
        tempMap.remove(messageId);
        log.info("消息已从待确认队列中移除, messageId=" + messageId);
    }

    // 从硬盘中读取全部数据，恢复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, ClassNotFoundException, MqException {
        // 先清空全部数据
        exchangeMap.clear();
        queueMap.clear();
        exchangeBindingMap.clear();
        queueBindingMap.clear();
        messageMap.clear();
        queueMessageMap.clear();

        // 恢复交换机数据
        List<Exchange> exchangeList = diskDataCenter.selectAllExchanges();
        for (Exchange exchange : exchangeList) exchangeMap.put(exchange.getName(), exchange);
        log.info("恢复 Exchange 数据完成");

        // 恢复队列数据
        List<MSGQueue> queueList = diskDataCenter.selectAllQueue();
        for (MSGQueue queue : queueList) queueMap.put(queue.getName(), queue);
        log.info("恢复 Queue 数据完成");

        // 恢复绑定数据
        List<Binding> bindingList = diskDataCenter.selectAllBinding();
        for (Binding binding : bindingList) {
            ConcurrentHashMap<String, Binding> queueMap = exchangeBindingMap.computeIfAbsent(binding.getExchangeName(),
                    k -> new ConcurrentHashMap<>());
            ConcurrentHashMap<String, Binding> exchangeMap = queueBindingMap.computeIfAbsent(binding.getQueueName(),
                    k -> new ConcurrentHashMap<>());

            queueMap.put(binding.getQueueName(), binding);
            exchangeMap.put(binding.getExchangeName(), binding);
        }
        log.info("恢复 Binding 数据完成");

        // 恢复消息数据（全部队列下的全部消息）
        for (MSGQueue queue : queueList) {
            LinkedList<Message> messageList = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            // 恢复到未取走的消息
            queueMessageMap.put(queue.getName(), messageList);
            // 恢复到消息中心
            for (Message message : messageList) messageMap.put(message.getMessageId(), message);
            /*
             * 此时不需要处理 waitAckMap，因为调用该方法时，表示消费者尝试与服务器重新建立连接，将全部消息设为未取走即可。
             * 而消息只有在取走后才涉及到应答，因此此处无需考虑应答。
             */
        }
    }
}
